package cn.linkai.hadoop.invertedindex;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.Reducer;
import org.apache.hadoop.mapreduce.lib.input.FileSplit;

import java.io.IOException;
import java.util.Iterator;
import java.util.StringTokenizer;

public class InvertedIndexTwiceMRFirst {

    public static class InvertedIndexTwiceMRFirstMapper extends Mapper<LongWritable,Text,Text,LongWritable>{

        private static LongWritable count=new LongWritable(1L);

        @Override
        protected void map(LongWritable key, Text value, Context context) throws IOException, InterruptedException {
            StringTokenizer tokenizer=new StringTokenizer(value.toString());
            String fileName=((FileSplit)context.getInputSplit()).getPath().toString();
            fileName=fileName.substring(fileName.lastIndexOf("/")+1,fileName.length());
            while(tokenizer.hasMoreTokens()){
                String word=tokenizer.nextToken();
                context.write(new Text(word+"->"+fileName),count);
            }
        }
    }

    public static class InvertedIndexTwiceMRFirstReducer extends Reducer<Text,LongWritable,Text,LongWritable>{
        @Override
        protected void reduce(Text key, Iterable<LongWritable> values, Context context) throws IOException, InterruptedException {
            long sum=0;
            Iterator<LongWritable> it=values.iterator();
            while(it.hasNext()){
                sum+=it.next().get();
            }
            context.write(key,new LongWritable(sum));
        }
    }
}
